package com.fei.kafka;

import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumerTest {

    private ConsumerConnector consumer;

    private KafkaConsumerTest() {
        Properties props = new Properties();
        //zookeeper 配置
        props.put("zookeeper.connect", "node1:2181/k_cluster1");

        //group 代表一个消费组
        props.put("group.id", "testGroup1");

        //zk连接超时
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");
        //序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");

        ConsumerConfig config = new ConsumerConfig(props);

        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
    }

    void consume(String name) {
    	new Thread(new ConsumerWork(consumer,name)).start();
    	
    }

    public static void main(String[] args) {
        new KafkaConsumerTest().consume(args[0]);
    }
}
